package com.example.bootproducer;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitOperations;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;


// Generated by https://start.springboot.io
// 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn

@SpringBootApplication
public class BootProducerApplication {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public static void main(String[] args) {
        SpringApplication.run(BootProducerApplication.class, args);
    }


    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        configurer.configure(rabbitTemplate, connectionFactory);


        //添加确认回调(生产端监听消息是否正常到达交换机)
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    //消息正常到达交换机
                    System.out.println("消息到达交换机...");
                }else{
                    //消息为正常到达交换机（消息生产端发送失败） -- 重试（发到哪个交换机+路由键是什么+消息内容）
                    String exchange = correlationData.getReturned().getExchange();
                    String routingKey = correlationData.getReturned().getRoutingKey();
                    Message message = correlationData.getReturned().getMessage();
                    //获取消息的唯一标识（可以利用这个唯一标识结合redis实现重发计数，达到限制后，添加到数据库，后续用定时任务重发）
                    String messageId = message.getMessageProperties().getMessageId();

                    rabbitTemplate.send(exchange,routingKey,message);
                }
            }
        });

        //返回回调(生产端监听消息是否正常到达队列)
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            //当消息到达队列失败后会回调
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                System.out.println("到达队列失败，消息重发............");
                Message message = returned.getMessage();
                String messageId = message.getMessageProperties().getMessageId();

                Long increment = redisTemplate.opsForValue().increment(messageId);
                redisTemplate.expire(messageId,10, TimeUnit.SECONDS);

                if (increment < 3) {
                    String exchange = returned.getExchange();
                    String routingKey = returned.getRoutingKey();

                    //实现重发
                    rabbitTemplate.send(exchange, routingKey, message);
                }else{
                    //表结构：交换机名称、路由键、消息、重试时间、重试次数、状态
                    System.out.println("记录到数据库的消息重试表中");
                }
            }
        });

        return rabbitTemplate;
    }
}
